package com.fwmagic.spark.core.cases.ser

import java.net.InetAddress
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext, TaskContext}

/**
  * 闭包现象：
  * 函数(eg：map)的内部用到了函数外部的引用类型，函数是在Executor端被调用的，
  * 在Driver端引用的数据要伴随着Task一起发送到Executor中
  *
  * 知识点：
  * 1、如果在Driver初始化了一个object或new一个class实例，然后在函数中使用，必须实现序列化接口。
  * 2、object在一个Executor中只有一份，因为object是一个单例对象，一个进程中只能有一个new的class时，
  * 一个Task独享一个实例，有几个task就有几个RulesMapClass实例。
  *
  *
  * 4种情况总结：
  * 1、定义个一个object单例类，实现序列化接口。在driver端初始化，并在函数中引用该对象。此时在每个Executor中只会有一份该实例对象
  * 2、定义一个普通的class类，实现序列化接口。在driver端初始化，并在函数中引用该对象，此时在每个Executor中的每个Task都会有一份实例对象
  * 3、定义个一个object单例类，不实现序列化接口，在函数内初始化，此时在每个Executor中只会有一份该实例对象
  * 4、定义一个普通的class类，不实现序列化接口，在函数内部初始化，每来一条数据new一个类的实例，这样会频繁执行，浪费大量资源。
  * 5、定义一个普通的class类，不实现序列化接口，在（mapPartitions）函数内部初始化，每一个分区共用一个RulesMapClassNoSer对象
  */
object SerTest3 {
    def main(args: Array[String]): Unit = {

        System.setProperty("HADOOP_USER_NAME","hadoop")

        val isLocal = args(0).toBoolean

        val conf: SparkConf = new SparkConf().setAppName(this.getClass.getSimpleName)
        if (isLocal) {
            conf.setMaster("local[*]")
        }

        val sc = new SparkContext(conf)

        val words: RDD[String] = sc.textFile(args(1))

        //封装数据
        val res = words.map(word => {
            //在executor端初始化不需要实现序列化接口，一个Executor中共用一个实例
            val rs = RulesMapObjNoSer
            val province: Option[String] = rs.rules.get(word)
            val partitionId: Int = TaskContext.getPartitionId()
            val hostName: String = InetAddress.getLocalHost.getHostName
            (word, province, partitionId, hostName, rs)
        })

        //将结果写到hdfs中
        res.saveAsTextFile(args(2))

        sc.stop()
    }

}
